package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

import static org.apache.flink.api.common.state.StateTtlConfig.UpdateType.OnCreateAndWrite;

/**
 * @Author sw
 * @CreateTime 2024-11-19
 * 12.2 流量域版本-渠道-地区-访客类别粒度页面浏览各窗口汇总表
 */
public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(10022,4,
                Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW,
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
        System.out.println("");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
//        dataStreamSource.print();
        //清洗数据
        SingleOutputStreamOperator<JSONObject> etlStream = etl(dataStreamSource);
//        etlStream.print();
        // key 之前添加水印和之后添加之后

        // 2. 添加水位线，并keyby  mid
        // 3. 状态编程(转成实体对象)-->求UV和PV
        SingleOutputStreamOperator<TrafficPageViewBean> keyedStream = keyby(etlStream);
        SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream = getWatermarksStream(keyedStream);
        // 4. 进行key ,key的字段vc ch ar is_new
        // 5. reduce累计在一起
        SingleOutputStreamOperator<TrafficPageViewBean> reduceStream = getReduce(watermarksStream);
        reduceStream.print();
        // 6. 写出到doris api
        writeToDoris(reduceStream);
    }

    private static void writeToDoris(SingleOutputStreamOperator<TrafficPageViewBean> reduceStream) {
        reduceStream.map(new DorisMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> getReduce(SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream) {
         return watermarksStream.keyBy(new KeySelector<TrafficPageViewBean, String>() {
                    @Override
                    public String getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                        return trafficPageViewBean.getVc() + ":" + trafficPageViewBean.getCh()
                                + ":" + trafficPageViewBean.getAr() + ":" + trafficPageViewBean.getIsNew();
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean x, TrafficPageViewBean y) throws Exception {
                                //滚动窗口累加不会有重叠，直接累加就可以，滑动会有重叠，需要重新创建新对象
                                x.setUvCt(x.getUvCt() + y.getUvCt());
                                x.setPvCt(x.getPvCt() + y.getPvCt());
                                x.setSvCt(x.getSvCt() + y.getSvCt());
                                x.setDurSum(x.getDurSum() + y.getDurSum());
                                return x;
                            }
                        },
                        new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                                long start = timeWindow.getStart();
                                long end = timeWindow.getEnd();
                                Iterator<TrafficPageViewBean> iterator = iterable.iterator();
                                TrafficPageViewBean trafficPageViewBean = iterator.next();
                                trafficPageViewBean.setStt(DateFormatUtil.tsToDate(start));
                                trafficPageViewBean.setEdt(DateFormatUtil.tsToDate(end));
                                trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                                collector.collect(trafficPageViewBean);
                            }
                        });
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> getWatermarksStream(SingleOutputStreamOperator<TrafficPageViewBean> keyedStream) {
        return keyedStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                                return trafficPageViewBean.getTs();
                            }
                        })
                .withIdleness(Duration.ofSeconds(120L)));
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> keyby(SingleOutputStreamOperator<JSONObject> etlStream) {
         return etlStream.keyBy(x -> x.getJSONObject("common").getString("mid"))
                .process(new ProcessFunction<JSONObject, TrafficPageViewBean>() {
                    ValueState<String> lastVisitedState;

                    @Override
                    public void open(Configuration parameters) {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                        //ttl设置状态过期时间
                        stateDescriptor.enableTimeToLive(StateTtlConfig
                                .newBuilder(org.apache.flink.api.common.time.Time.hours(24))
                                //默认更新
                                .setUpdateType(OnCreateAndWrite).build());
                        lastVisitedState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject,
                                               ProcessFunction<JSONObject, TrafficPageViewBean>.Context context,
                                               Collector<TrafficPageViewBean> collector) throws Exception {
                        Long ts = jsonObject.getLong("ts");
                        String curDt = DateFormatUtil.tsToDate(ts);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject page = jsonObject.getJSONObject("page");
                        String lastValue = lastVisitedState.value();
                        Long uv = 0L;
                        Long pv = 1L;
                        Long sv = 0L;
                        //当天第一次登录
                        if (!curDt.equals(lastValue)) {
                            uv = 1L;
                            //更新状态
                            lastVisitedState.update(curDt);
                        }

                        //上一页未访问任何地址，即为刚进入算作一次会话数
                        String lastPageId = page.getString("last_page_id");
                        if (lastPageId == null) {
                            sv = 1L;
                        }

                        collector.collect(TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .isNew(common.getString("is_new"))
                                .uvCt(uv)
                                .svCt(sv)
                                .pvCt(pv)
                                .durSum(page.getLong("during_time"))
                                .ts(ts)
                                .sid(common.getString("sid"))
                                .build());
                    }
                });
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                JSONObject page = jsonObject.getJSONObject("page");
                JSONObject common = jsonObject.getJSONObject("common");
                String ts = jsonObject.getString("ts");
                if (common != null && page != null) {
                    String mid = common.getString("mid");
                    if (mid != null && ts != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });
    }

}
